import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class ZuiDaZhi {
    public static class ZuiDaZhimap extends Mapper<LongWritable, Text, Text, PaiXu> {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String s = value.toString();
            String[] line = s.split(",");
            PaiXu p = new PaiXu();
            p.setOrderId(line[0]);
            p.setUserId(line[1]);
            p.setName(line[2]);
            p.setNumber(Double.parseDouble(line[3]) * Double.parseDouble(line[4]));
            context.write(new Text(line[0]), p);
        }
    }

    public static class ZuiDaZhiReduce extends Reducer<Text, PaiXu, Text, PaiXu> {
        @Override
        protected void reduce(Text key, Iterable<PaiXu> values, Context context) throws IOException, InterruptedException {
            List<PaiXu> paiXuList = new ArrayList<PaiXu>();
            for (PaiXu value : values) {
                PaiXu paiXu = new PaiXu();
                paiXu.setOrderId(value.getOrderId());
                paiXu.setUserId(value.getUserId());
                paiXu.setName(value.getName());
                paiXu.setNumber(value.getNumber());
                paiXuList.add(paiXu);
            }
            Collections.sort(paiXuList);
            for (int i = 0; i < paiXuList.size(); i++) {
                if (i == 3) {
                    break;
                }
                context.write(new Text(key), paiXuList.get(i));
            }
        }


        public static void main(String[] args) {
            Configuration conf = new Configuration();

            try {
                //新建一个Job工作
                Job job = new Job(conf);
                //设置运行类
                job.setJarByClass(ZuiDaZhi.class);
                //设置要执行的mapper类（自己书写的）
                job.setMapperClass(ZuiDaZhimap.class);
                //设置要执行的reduce类（自己书写的）
                job.setReducerClass(ZuiDaZhiReduce.class);
                //设置输出key的类型
                job.setMapOutputKeyClass(Text.class);
                //设置输出value的类型
                job.setMapOutputValueClass(PaiXu.class);

                File file = new File("d:/file/output");
                if (file.exists()) {
                    FileUtil.fullyDelete(file);
                }
                //mapreduce 输入数据的文件/目录
                FileInputFormat.addInputPath(job, new Path("d:/file/input"));
                //mapreduce 执行后输出的数据目录
                FileOutputFormat.setOutputPath(job, new Path("d:/file/output"));
                //设置ruduce任务的个数，默认个数为一个（一般reduce的个数越多效率越高）
                job.setNumReduceTasks(1);
                //执行完毕退出
                System.exit(job.waitForCompletion(true) ? 0 : 1);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }


    }
}